package com.loong.data.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.loong.common.enumerate.OrderStatus;
import com.loong.common.rabbitmq.consumer.BaseConsumer;
import com.loong.common.rabbitmq.dto.OrderDTO;
import com.loong.common.rabbitmq.dto.base.CommonEventDto;
import com.loong.data.handler.OrderHandler;
import com.loong.data.model.OrderModel;
import com.loong.global.constant.MQConstant;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Description: es 订单 消费
 * @Author: hzp
 * @Date: 2020-12-29 14:53
 */
@Slf4j
@Component
public class OrderReportConsumer extends BaseConsumer  /*implements BatchMessageListener*/ {
   // @RabbitListener(queues = {MQConstant.BASE_PREFIX + MQConstant.QUEUE_NAME.ORDER_ES + MQConstant.BASE_SUFFIX.QUEUE}, containerFactory = "batchQueueRabbitListenerContainerFactory")
//    @Override
//    public void onMessageBatch(List<Message> messages) {
//        log.info("batch.queue.consumer ES Topic 队列收到{}条message", messages.size());
//        if(messages.size()>0){
//            log.info("第一条数据是: {}", new String(messages.get(0).getBody()));
//        }
//    }

    private static Long nowSaveStaticTime = 0L ;
    private static Long nowUpdateStaticTime = 0L ;
    private static Long nowDeleteStaticTime = 0L ;
    private static Long deadNowStaticTime;
    private static final List<OrderDTO> saveDataList = new ArrayList<>();
    private static final   List<OrderDTO> updateDataList = new ArrayList<>();
    private static final List<OrderDTO> deleteDataList = new ArrayList<>();
    private static final List<OrderDTO> dataDeadList = new ArrayList<>();
    /**
     * 用于控制 一次性操作的集合大小 暂时写这里测试
     */
    private static final int size = 20;
    /**
     * 用于不管有没有消息 submitTime 时间到了就消费到es 暂时写这里测试
     */
    private static final int submitTime = 5;

    @Resource
    private OrderHandler orderHandler;

    /**
     * 监听es 订单延迟队列
     * @param channel 通道
     * @param message 消息
     * @throws IOException IOException
     * @throws InterruptedException InterruptedException
     */
    @RabbitListener(queues = {MQConstant.BASE_PREFIX + MQConstant.QUEUE_NAME.ORDER_ES + MQConstant.BASE_SUFFIX.QUEUE}, containerFactory = "defaultRabbitListener")
    public void receiveEsEvent(Channel channel, Message message) throws IOException, InterruptedException {
        log.debug("监听到====ES Topic 延迟队列======");
       // long deliveryTag = message.getMessageProperties().getDeliveryTag();
        byte[] data = message.getBody();
        CommonEventDto<JSONObject> eventDto = JSON.parseObject(new String(data)).toJavaObject(CommonEventDto.class);
        Short eventCode = eventDto.getEventCode();
        OrderDTO orderDTO = eventDto.getData().toJavaObject(OrderDTO.class);
        log.debug("====ES Topic orderDTO=====:{}",orderDTO);
        //本地缓存 订单信息
        addDataList(eventCode,orderDTO);
        //消费缓存的订单信息
        checkTimeAndExecutor();
        //消息确认放在 入库操作 入库完成 状态已支付了进行确认

    }

    /**
     * 处理N分钟后 未支付 的订单
     * @param channel 通道
     * @param message 消息
     * @throws IOException IO异常
     */
    @RabbitListener(queues = {MQConstant.BASE_PREFIX + MQConstant.QUEUE_NAME.ORDER_ES + MQConstant.BASE_SUFFIX.DEAD_QUEUE}, containerFactory = "defaultRabbitListener")
    public void receiveEsDeadEvent(Channel channel, Message message) throws IOException {
        log.debug("监听到====ES Topic 死信队列======");
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        byte[] data = message.getBody();
        CommonEventDto<JSONObject> eventDto = JSON.parseObject(new String(data)).toJavaObject(CommonEventDto.class);
        OrderDTO orderDTO = eventDto.getData().toJavaObject(OrderDTO.class);
        log.debug("====ES Topic orderDTO=====:{}",orderDTO);
        OrderModel esModel = orderHandler.findOrderModelById(orderDTO.getOrderBase().getId());
        //未支付 进行删除es中订单
        if(null!=esModel&&(OrderStatus.WAITING_PAYMENT.getCode()==esModel.getStatus())){
            //删除es 中的订单
            orderHandler.deleteById(esModel.getId());
        }
        submitAck(channel, deliveryTag);

    }
    private void addDataList(short eventCode,OrderDTO orderDTO){
        switch (eventCode){
            case 1:
                //save
                saveDataList.add(orderDTO);
                break;
            case 2:
                //update
                updateDataList.add(orderDTO);

                break;
            case 3:
                //delete
                deleteDataList.add(orderDTO);
                break;
        }
    }

    //xxl Job
    public void checkTimeAndExecutor(){
        log.debug("rpc 进入");
        log.debug("时间 :{} data: {}",nowSaveStaticTime,saveDataList);
        long nowTime = System.currentTimeMillis();

                if(nowSaveStaticTime ==0L){
                    nowSaveStaticTime = nowTime;
                }
                if((nowTime - nowSaveStaticTime)/1000>submitTime&&saveDataList.size()>0){
                    eventExecutor("save");
                }else if(saveDataList.size()>=size){
                    eventExecutor("save");
                }

                //update
                if(nowUpdateStaticTime ==0){
                    nowUpdateStaticTime = nowTime;
                }
                if((nowTime - nowUpdateStaticTime)/1000>submitTime&&updateDataList.size()>0){
                    eventExecutor("update");
                }else if(updateDataList.size()>=size){
                    eventExecutor("update");
                }


                //delete
                if(nowDeleteStaticTime ==0){
                    nowDeleteStaticTime = nowTime;
                }
                if((nowTime - nowDeleteStaticTime)/1000>submitTime&&deleteDataList.size()>0){
                    eventExecutor("delete");
                }else if(deleteDataList.size()>=size){
                    eventExecutor("delete");
                }


    }


    private void eventExecutor(String type){
        switch (type){
            case "save":
                //save
                orderHandler.saveOrderModel(saveDataList);
                saveDataList.clear();
                nowSaveStaticTime = 0L;
                break;
            case "update":
                //update
                orderHandler.updateOrderModel(updateDataList);
                updateDataList.clear();
                nowUpdateStaticTime = 0L;
                break;
            case "delete":
                //delete
                orderHandler.deleteShopModel(deleteDataList);
                deleteDataList.clear();
                nowDeleteStaticTime = 0L;
                break;
        }
    }
}
